package com.jh.spring.cloud.stream.binder.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.util.List;

/**
 * @author Administrator
 * @projectName newforward
 * @title
 * @package com.jh.spring.cloud.stream.binder.rocketmq
 * @description
 * @date 2019/11/3 11:38
 */
public class RocketMQMessageChannelBinder implements Binder<MessageChannel, ConsumerProperties, ProducerProperties> {

    private final String groupName = "default_group";
    private final String nameSrvAddr = "localhost:9876";
    private final String topic = "default_topic";
    private final String tag = "default_tag";


    @Override
    @SuppressWarnings("all")
    public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel inputChannel, ConsumerProperties consumerProperties) {

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        // Specify name server addresses.
        consumer.setNamesrvAddr(nameSrvAddr);

        try {

            // Subscribe one more more topics to consume.
            consumer.subscribe(topic, tag);

            // Register callback to execute on arrival of messages fetched from brokers.
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                    System.out.printf("接收消息: %s %n", Thread.currentThread().getName(), msgs);

                    msgs.forEach(msg -> {
                        byte[] body = msg.getBody();
                        // 发送消息到消息管道
                        inputChannel.send(new GenericMessage<>(body));
                    });

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            //Launch the consumer instance.
            consumer.start();
            System.out.printf("Consumer Started.%n");

        } catch (Exception e) {
            e.printStackTrace();
        }

        return () -> {
            System.out.println("Consumer Shutdown");
            /**
             * 当ApplicationContext关闭的时候进行回调
             */
            consumer.shutdown();
        };
    }

    @Override
    @SuppressWarnings("all")

    public Binding<MessageChannel> bindProducer(String name, MessageChannel outputChannel, ProducerProperties producerProperties) {

        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer(groupName);
        // Specify name server addresses.
        producer.setNamesrvAddr(nameSrvAddr);
        try {
            //Launch the instance.
            producer.start();

            // 消息订阅回调
            SubscribableChannel channel = (SubscribableChannel) outputChannel;
            channel.subscribe(message -> {

                try {
                    Object msgBody = message.getPayload();

                    Message mqMsg = new Message();
                    mqMsg.setTopic(topic);
                    mqMsg.setTags(tag);
                    mqMsg.setBody(this.changePayload2Body(msgBody));

                    SendResult sendResult = producer.send(mqMsg);
                    System.out.printf("消息发送：%s%n", sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });

        } catch (Exception e) {
            e.printStackTrace();
        }

        return () -> {
            System.out.println("Producer Shutdown");
            producer.shutdown();
        };
    }

    private byte[] changePayload2Body(Object payload) throws Exception {

        if (payload instanceof byte[]) {
            return (byte[]) payload;
        }

        // Object 写入字节流
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(payload);
        return bos.toByteArray();
    }
}
